package com.hx.thread.async;

import com.google.common.base.Preconditions;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * java批量请求
 *
 * @author MinghuaChen
 * @since 2022/8/20 下午5:02
 */
public class BatchRequestApplication {

    private static final Logger logger = LoggerFactory.getLogger(BatchRequestApplication.class);

    public static void main(String[] args) {
        List<String> skuIdList = Arrays.asList(
                "S1", "S4", "S3", "S4", "S5", "S6", "S7", "S8", "S9"
        );

        Stopwatch stopwatch = Stopwatch.createStarted();

        logger.info("start batch request .....");

        // 默认10个一批次
        int defaultSize = 2;

        List<String> resultList = partitionCall2ListAsync(skuIdList, defaultSize, null, BatchRequestApplication::call);
        long millisecond = stopwatch.stop().elapsed(TimeUnit.MILLISECONDS);

        logger.info("end batch request ,cost time: {}ms", millisecond);

        resultList.forEach(System.out::println);

    }


    /**
     * @param requestDataList 请求数据
     * @param size            分片因子
     * @param executorService 结果
     * @param function        调用函数
     * @param <T>             请求如参范型
     * @param <V>             出参范型
     * @return 响应结果
     */
    public static <T, V> List<V> partitionCall2ListAsync(
            List<T> requestDataList,
            int size,
            ExecutorService executorService,
            Function<List<T>, List<V>> function) {
        if (requestDataList == null || requestDataList.size() == 0) {
            return new ArrayList<>(0);
        }
        Preconditions.checkArgument(size > 0, "size must not be minus");
        // 异步请求
        List<CompletableFuture<List<V>>> completableFutureList = Lists.partition(requestDataList, size)
                .stream()
                .map(dataList -> {
                    if (executorService == null) {
                        return CompletableFuture.supplyAsync(() -> function.apply(dataList))
                                .exceptionally(e -> {
                                    logger.error("执行获取结果异常", e);
                                    return new ArrayList<>();
                                });
                    } else {
                        return CompletableFuture.supplyAsync(() -> function.apply(dataList), executorService)
                                .exceptionally(e -> {
                                    logger.error("执行获取结果异常", e);
                                    return new ArrayList<>();
                                });
                    }
                })
                .collect(Collectors.toList());
        // 等待全部返回
        CompletableFuture<Void> completableFuture = CompletableFuture.allOf(completableFutureList.toArray(new CompletableFuture[0]));
        try {
            // ack the acquire result time
            completableFuture.get(1, TimeUnit.SECONDS);
        } catch (TimeoutException | InterruptedException | ExecutionException exception) {
            logger.error("acquire result occur exception",exception);
            throw new RuntimeException();
        }
        return completableFutureList.stream()
                // 获取结果
                .map(CompletableFuture::join)
                // 过滤结果为空的
                .filter(e -> e.size() != 0)
                // 结果归并
                .reduce(new ArrayList<V>(), ((list1, list2) -> {
                    List<V> resultList = new ArrayList<>();
                    if (list1 != null && list1.size() > 0) {
                        resultList.addAll(list1);
                    }
                    if (list2 != null && list2.size() > 0) {
                        resultList.addAll(list2);
                    }
                    return resultList;
                }));
    }


    /**
     * 调用函数
     *
     * @param skuIdList skuId列表
     * @return 加工后的skuId列表
     */
    public static List<String> call(List<String> skuIdList) {
        if (skuIdList == null || skuIdList.size() == 0) {
            return new ArrayList<>();
        }
        logger.info("process request");
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (Exception exception) {
            throw new RuntimeException("运行出错");
        }
        if (skuIdList.contains("S99")) {
            throw new RuntimeException("S5刻意制造异常");
        }
        return skuIdList.stream().map(e -> e.concat("xxxx")).collect(Collectors.toList());
    }

}
